use std::sync::{Arc, atomic::AtomicU16};

use arrow::{array::*, datatypes::*};
use chrono::Timelike;
use itertools::izip;
use parquet::arrow::arrow_reader::ParquetRecordBatchReader;
use rayon::iter::{IntoParallelRefIterator, ParallelIterator};

fn main() {
    let instant = std::time::Instant::now();
    let files = std::fs::read_dir("../data/data")
        .unwrap()
        .into_iter()
        .filter(|x| {
            let x = x.as_ref().unwrap();
            let x = x.file_name();
            let x = x.into_string().unwrap();
            x.starts_with("2020-01")
        })
        .map(|x| {
            let i = x.unwrap();
            let i = i.path();
            i
        })
        .collect::<Vec<_>>();
    let count = AtomicU16::new(0);
    let data = files
        .par_iter()
        .map(|x| {
            count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            println!("{:?}", count);
            let x = std::fs::File::open(x).unwrap();
            let data = ParquetRecordBatchReader::try_new(x, 1024 * 1024).unwrap();
            let mut h11 = hashbrown::HashMap::new();
            let mut h22 = hashbrown::HashMap::new();
            let mut h33 = hashbrown::HashMap::new();
            for i in data {
                let i = i.unwrap();
                let dan_wei = i.column(0).as_string::<i64>();
                let saler = i.column(1).as_string::<i64>();
                let sale_time = i.column(2).as_primitive::<TimestampMillisecondType>();
                let sale_good = i.column(3).as_string::<i64>();
                let sale_num = i.column(4).as_primitive::<UInt32Type>();
                let mut h1 = hashbrown::HashMap::new();
                let mut h2 = hashbrown::HashMap::new();
                let mut h3 = hashbrown::HashMap::new();
                for (dan_wei, saler, sale_time, sale_good, sale_num) in
                    izip!(dan_wei, saler, sale_time, sale_good, sale_num)
                {
                    let tmp = chrono::DateTime::from_timestamp_millis(sale_time.unwrap()).unwrap().naive_local();
                    let tmpp = h1.entry((dan_wei.unwrap(), tmp.hour())).or_insert(0);
                    *tmpp += sale_num.unwrap();
                    if tmp.hour() == 12 {
                        let tmpp = h2
                            .entry((dan_wei.unwrap(), sale_good.unwrap()))
                            .or_insert(hashbrown::HashSet::new());
                        tmpp.insert(saler.unwrap().to_string());
                        let tmpp = h3.entry((dan_wei.unwrap(), saler.unwrap())).or_insert(0);
                        *tmpp += sale_num.unwrap()
                    }
                }
                for i in h1 {
                    let tmp = h11.entry((i.0.0.to_string(), i.0.1)).or_insert(0);
                    *tmp += i.1
                }
                for i in h2 {
                    let tmp = h22
                        .entry((i.0.0.to_string(), i.0.1.to_string()))
                        .or_insert(hashbrown::HashSet::new());
                    for j in i.1 {
                        tmp.insert(j);
                    }
                }
                for i in h3 {
                    let tmp = h33
                        .entry((i.0.0.to_string(), i.0.1.to_string()))
                        .or_insert(0);
                    *tmp += i.1
                }
            }
            let mut h111 = hashbrown::HashMap::new();
            let mut h222 = hashbrown::HashMap::new();
            let mut h333 = hashbrown::HashMap::new();
            for i in h11 {
                let tmp = h111.entry(i.0.0).or_insert((0, 0));
                *tmp = (i.1 + tmp.0, tmp.1 + 1)
            }
            for i in h22 {
                let tmp = h222.entry(i.0.0).or_insert((0, 0));
                *tmp = (i.1.len() + tmp.0, tmp.1 + 1)
            }
            for i in h33 {
                let tmp = h333.entry(i.0.0).or_insert((0, 0));
                *tmp = (i.1 + tmp.0, tmp.1 + 1)
            }
            let mut hh1=hashbrown::HashMap::new();
            let mut hh2=hashbrown::HashMap::new();
            let mut hh3=hashbrown::HashMap::new();
            for i in h111{
                hh1.insert(i.0, i.1.0 as f32/i.1.1 as f32);
            }
            for i in h222{
                hh2.insert(i.0, i.1.0 as f32/i.1.1 as f32);
            }
            for i in h333{
                hh3.insert(i.0, i.1.0 as f32/i.1.1 as f32);
            }
            let mut daily_out = hashbrown::HashMap::new();
            for i in hh1{
                daily_out.insert(i.0.clone(), (i.1,hh2.get(&i.0).unwrap_or(&0.).to_owned(),hh3.get(&i.0).unwrap_or(&0.).to_owned()));
            }
            daily_out
        })
        .collect::<Vec<_>>();
    let mut data1 = hashbrown::HashMap::new();
    for i in data {
        for j in i {
            let tmp = data1
                .entry(j.0)
                .or_insert((0.0, 0, 0.0, 0, 0.0, 0));
            *tmp = (
                j.1.0 + tmp.0,
                tmp.1 + 1,
                j.1.1 + tmp.2,
                tmp.3 + 1,
                j.1.2 + tmp.4,
                tmp.5 + 1,
            );
        }
    }
    let mut data2 = hashbrown::HashMap::new();
    for i in data1 {
        data2.insert(
            i.0,
            (
                i.1.0 as f32 / i.1.1 as f32,
                i.1.1,
                i.1.2 as f32 / i.1.3 as f32,
                i.1.3,
                i.1.4 as f32 / i.1.5 as f32,
                i.1.5,
            ),
        );
    }
    println!("{:?},{}", data2, instant.elapsed().as_secs_f32());
}
